package cn._51doit.akka.rpc

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.collection.mutable
import scala.concurrent.duration._
/**
 * @create: 2021-10-08 19:41
 * @author: 今晚打脑斧
 * @program: Master
 * @Description:
 *    Master
 **/
class Master extends Actor{
  val id2Worker = new mutable.HashMap[String,WorkerInfo]()

  override def preStart(): Unit = {
    //启动一个定时器 定期检测超时的worker
    import context.dispatcher
    context.system.scheduler.schedule(0 second, 6 second, self, CheckTimeOutWorker)
  }

  //接收消息的方法
  override def receive: Receive = {
    case RegisterWorker(workerId, memory, cores) => {
      //将Worker的信息保存起来
      val workerInfo = new WorkerInfo(workerId, memory, cores)
      //将workerInfo保存到Map中
      id2Worker(workerId) = workerInfo
      //向Worker发送一个注册成功的消息
      sender() ! RegisteredWorker
    }

      case HeartBeat(workerId) => {
        println("receive a heartbeat msg from worker: " + workerId)
        //更新对应Worker的最近一次心跳时间
        if(id2Worker.contains(workerId)) {
          val workerInfo = id2Worker(workerId)
          //将当前的系统时间赋给对应WorkerInfo的最近的心跳时间
          workerInfo.lastHeartBeatTime = System.currentTimeMillis();
        }
      }

      case CheckTimeOutWorker => {
        //检测超时的Worker
        val deadWorkers = id2Worker.values.filter(w => System.currentTimeMillis() - w.lastHeartBeatTime > 15000)
        //将超时的Worker从Map中移除
        deadWorkers.foreach(w => id2Worker -= w.workerId)
        println(s"current alive worker size is : ${id2Worker.size}")
      }
  }
}
object Master{
  val MASTER_ACTOR_SYSTEM_NAME = "MasterActorSystem"
  val MASTER_ACTOR_NAME = "MasterActor"
  def main(args: Array[String]): Unit = {

    val host = args(0)      //Master的ip地址
    val port = args(1).toInt//Master的端口号

    val configStr =
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = $host
        |akka.remote.netty.tcp.port = $port
        |""".stripMargin;
    val config = ConfigFactory.parseString(configStr);
    //创建ActorSystem(单例的)
    val masterActorSystem = ActorSystem(MASTER_ACTOR_SYSTEM_NAME, config)
    //创建Actor  通过actorof创建对应的Actor
    masterActorSystem.actorOf(Props[Master], MASTER_ACTOR_NAME)
//    val masterActor = masterActorSystem.actorOf(Props[Master], "MasterActor")
//    //发送消息
//    masterActor ! "hi" ; //发送异步消息
//    /*
//    1 + 1
//    1 .+ (1)
//     */
  }
}
